package com.smsc.headend.connector.manager;


import cn.hutool.core.date.SystemClock;
import cn.hutool.json.JSONUtil;
import com.google.protobuf.InvalidProtocolBufferException;
import com.smsc.headend.common.utils.ProtoBufferUtil;
import com.smsc.headend.connector.bo.ConnectTaskResultDto;
import com.smsc.headend.connector.service.DeviceConnectService;
import com.smsc.headend.connector.service.TaskService;
import com.smsc.headend.connector.service.TerminalChannelService;
import com.smsc.headend.module.asset.entity.ComEntity;
import com.smsc.headend.module.protocol.cosem.util.ByteUtils;
import com.smsc.headend.module.task.dto.ConnectorTask;
import com.smsc.headend.module.task.dto.ConnectorTaskResult;
import com.smsc.headend.module.task.proto.ConnectorTaskProto;
import io.micrometer.core.instrument.Counter;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang.time.DateFormatUtils;
import org.apache.commons.lang.time.DateUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.kafka.config.KafkaListenerEndpointRegistry;
import org.springframework.kafka.listener.MessageListenerContainer;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.Date;
import java.util.Optional;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

/**
 * 接收UTE发送来的ConnectorTask，通过此runner执行发送到设备
 */
@Slf4j
public class ConnectorTaskThread implements Runnable {

    ConnectorTask connectorTask;

    ComEntity terminal;

    Integer ticketNumber;

    @Autowired
    DeviceConnectService deviceConnectService;

    @Autowired
    TaskService taskService;

    @Autowired
    @Qualifier("taskInboundCounter")
    Counter taskCounter;

    @Autowired
    TaskTicketManager taskTicketManager;

    public ConnectorTaskThread(ConnectorTask connectorTask, ComEntity terminal, Integer ticket) {
        this.connectorTask = connectorTask;
        this.terminal = terminal;
        this.ticketNumber = ticket;
    }

    public ConnectorTask getConnectorTask() {
        return connectorTask;
    }

    public ComEntity getTerminal() {
        return terminal;
    }

    public Integer getTicketNumber() {
        return ticketNumber;
    }

    @Override
    public void run() {
        try {
            if (this.ticketNumber == null) {
                log.error("getTicketFailed TaskFailed,comId:{},taskNo:{},sendTime:{},ticket={}", connectorTask.getTerminalId(), connectorTask.getTaskNo(), DateFormatUtils.format(new Date(connectorTask.getSendTime()), DateFormatUtils.ISO_DATETIME_TIME_ZONE_FORMAT.getPattern()), this.ticketNumber);
                deviceConnectService.sendDeviceResultMessage(new ConnectTaskResultDto(ConnectorTaskResult.ExecuteCode.fail, connectorTask.getAtomicTaskNo(), null, "CON-module busy", connectorTask.getTerminalId(), connectorTask.getFlag(), this.ticketNumber), terminal);
                return;
            }
            log.debug("receive comId={}, task {} , a-task {}", connectorTask.getTerminalId(), connectorTask.getTaskNo(), connectorTask.getAtomicTaskNo());
            log.debug("packet info: a-task={} , packets={}", connectorTask.getAtomicTaskNo()
                    , connectorTask.getPacketBytes() == null ? null : JSONUtil.toJsonStr(Optional.ofNullable(Arrays.stream(connectorTask.getPacketBytes()).map(ByteUtils::byteToHexString).collect(Collectors.toList())).orElse(new ArrayList<>())));
            connectorTask.setTicketNumber(this.ticketNumber);
            try {
                taskCounter.increment();
                log.debug("TaskAccept:taskNo={},comId={},flag={},ticket={}", connectorTask.getTaskNo(), connectorTask.getTerminalId(), connectorTask.getFlag(), connectorTask.getTicketNumber());
                //设备为客户端模式，需要获取分布式锁
                if (ComEntity.CommunicationMode.CLIENT.getCode() == terminal.getComMode()) {
                    boolean addSuccess = taskService.addAtomicTask(connectorTask);
                    if (!addSuccess) {
                        log.debug("{},serverModeComTaskAcquireLockFailed", connectorTask.getAtomicTaskNo());
                        return;
                    }
                }
                deviceConnectService.sendToDevice(connectorTask);
            } catch (Exception e) {
                log.error("sendError", e);
                deviceConnectService.sendDeviceResultMessage(new ConnectTaskResultDto(ConnectorTaskResult.ExecuteCode.fail, connectorTask.getAtomicTaskNo(), null, "CON-error", connectorTask.getTerminalId(), connectorTask.getFlag(), this.ticketNumber), terminal);
            }
        } catch (Exception e) {
            log.error("unexpected error", e);
            deviceConnectService.sendDeviceResultMessage(new ConnectTaskResultDto(ConnectorTaskResult.ExecuteCode.fail, connectorTask.getAtomicTaskNo(), null, "CON-error", connectorTask.getTerminalId(), connectorTask.getFlag(), this.ticketNumber), terminal);
        }
    }
}
